// 源文件
#include"threadpool.h"
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>

const int NUMBER = 2;

ThreadPool* threadPoolCreate(int min, int max, int queueCapacity){

    /*
    * 用do{}while(0)可以很方便的释放内存，如果直接用流式，那么中途失败，则需要把之前的内存都释放；
    * 用do则可以先break，然后再统一释放
    */

    ThreadPool* pool = (ThreadPool*)malloc(sizeof(ThreadPool));
    do{
        // 开辟内存失败，则返回
        if (pool == NULL) {
            break;
        }

        pool->threaIDs = (pthread_t*)malloc(sizeof(pthread_t) * max);                       // 开辟工作者线程ID数组
        if (pool->threaIDs == NULL) {
            break;
        }

        // 参数赋初值
        memset(pool->threaIDs, 0, sizeof(pthread_t) * max);                                 // 给线程ID初值为0
        pool->maxNum = max;
        pool->minNum = min;
        pool->busyNum = 0;
        pool->liveNum = min;                                                                // 最开始存活的线程数就是min个
        pool->exitNum = 0;

        // 初始化互斥量和条件变量
        if (pthread_mutex_init(&pool->mutexPool, NULL) != 0 ||
            pthread_mutex_init(&pool->mutexBusy, NULL) != 0 ||
            pthread_cond_init(&pool->notEmpty, NULL) != 0   ||
            pthread_cond_init(&pool->notFull, NULL)) 
            {
            break;
        }

        // 任务队列开辟内存，初始化
        pool->taskQ = (Task*)malloc(sizeof(Task) * queueCapacity);
        pool->queueCapacity = queueCapacity;
        pool->queueSize = 0;
        pool->queueFront = 0;
        pool->queueRear = 0;

        // 启停标志
        pool->shutdown = 0;

        // 创建管理者线程
        pthread_create(&pool->managerID, NULL, manager, pool);                         // manager表示管理者线程函数名
        
        // 创建工作者线程
        for (int i = 0; i < min; ++i) {
            pthread_create(&pool->threaIDs[i], NULL, worker, pool);                   // worker表示工作者的线程函数名
        }
        return pool;

    }while(0);

    // 回收空间资源
    if (pool && pool->threaIDs) {
        free(pool->threaIDs);
    }
    if (pool && pool->taskQ) {
        free(pool->taskQ);
    }
    if (pool) {
        free(pool);
    }

    return NULL;
}

void* worker(void* args) {

    ThreadPool* pool = (ThreadPool*)args;

    // 循环一直读任务队列
    while (1) {

        // 对线程池加锁
        pthread_mutex_lock(&pool->mutexPool);

        // 任务队列为空
        while (pool->queueSize == 0 && !pool->shutdown) {
            // 阻塞工作线程，等待唤醒
            pthread_cond_wait(&pool->notEmpty, &pool->mutexPool);
            // 判断当前线程是否需要被销毁，销毁一个线程，则存活的线程数减一
            if (pool->exitNum > 0) {
                pool->exitNum--;
                if (pool->liveNum > pool->minNum) {
                    pool->liveNum--;
                }
                pthread_mutex_unlock(&pool->mutexPool);
                //pthread_exit(NULL);
                threadExit(pool);
            }
        }

        // 线程池是否关闭
        if (pool->shutdown) {
            // 解锁
            pthread_mutex_unlock(&pool->mutexPool);
            //pthread_exit(NULL);
            threadExit(pool);
        }

        // 从任务队列的头部取出一个任务
        Task task;
        task.function = pool->taskQ[pool->queueFront].function;
        task.args = pool->taskQ[pool->queueFront].args;

        // 移动队列头部索引，这里比较巧妙，“加1取余，可以保证索引后移，而且会在到达数组边界时，再回到初始值0”
        pool->queueFront = (pool->queueFront + 1) % pool->queueCapacity;            // 环形队列移动队头
        pool->queueSize--;

        pthread_mutex_unlock(&pool->mutexPool);
        pthread_cond_signal(&pool->notFull);        // 任务取出后，唤醒生产者线程

        printf("thread %ld start working!\n", pthread_self());

        // 工作线程+1，执行工作线程
        pthread_mutex_lock(&pool->mutexBusy);
        pool->busyNum++;
        pthread_mutex_unlock(&pool->mutexBusy);
        task.function(task.args);                   // 函数调用
        free(task.args);                            // 参数存放在堆区
        task.args = NULL;

        printf("thread %ld exit!\n", pthread_self());

        // 工作线程数-1
        pthread_mutex_lock(&pool->mutexBusy);
        pool->busyNum--;
        pthread_mutex_unlock(&pool->mutexBusy);
    }
    return NULL;
}


void* manager(void* args) {

    ThreadPool* pool = (ThreadPool*) args;
    
    while (!pool->shutdown) {
        // 每隔3秒检测一次
        sleep(3);

        // 取出线程池中的任务数量和当前线程数量
        pthread_mutex_lock(&pool->mutexPool);
        int queueSize = pool->queueSize;
        int liveNum = pool->liveNum;
        pthread_mutex_unlock(&pool->mutexPool);

        // 取出忙线程的数量
        pthread_mutex_lock(&pool->mutexBusy);
        int busyNum = pool->busyNum;
        pthread_mutex_unlock(&pool->mutexBusy);

        // 添加线程，规则为：
        // 任务的个数>存活的线程个数 && 存活的线程数<最大线程数
        if (queueSize > liveNum && liveNum < pool->maxNum) {
            pthread_mutex_lock(&pool->mutexPool);
            int counter = 0;
            // 从线程ID数组中找一个空闲的位置存放新建的线程ID，整个for循环只创建了NUMBER个线程
            for (int i = 0; i < pool->maxNum && counter < NUMBER && pool->liveNum < pool->maxNum; ++i) {
                if (pool->threaIDs[i] == 0) {
                    pthread_create(&pool->threaIDs[i], NULL, worker, pool);
                    counter++;
                    pool->liveNum++;
                }
            }
            pthread_mutex_unlock(&pool->mutexPool);
        }

        // 销毁线程，规则为：
        // 忙的线程*2 < 存活的线程数 && 存活的线程数 > 最小线程数
        if (busyNum * 2 < liveNum && liveNum > pool->minNum) {
            pthread_mutex_lock(&pool->mutexPool);
            pool->exitNum = NUMBER;
            pthread_mutex_unlock(&pool->mutexPool);
            // 引导线程自己销毁
            for (int i = 0; i < NUMBER; ++i) {
                // 唤醒阻塞的worker线程
                pthread_cond_signal(&pool->notEmpty);
            }
        }

    }
    return NULL;
}


void threadExit(ThreadPool* pool) {
    pthread_t tid = pthread_self();
    // 将线程ID数组对应的ID置为0，即回收数组位置
    for (int i = 0; i < pool->maxNum; ++i) {
        if (pool->threaIDs[i] == tid) {
            pool->threaIDs[i] = 0;
            break;
        }
    }
    pthread_exit(NULL);
}

void threadPoolAdd(ThreadPool* pool, void(*func)(void*), void* arg) {
    pthread_mutex_lock(&pool->mutexPool);

    // 如果当前线程池任务队列已经满了
    while (pool->queueSize == pool->queueCapacity && !pool->shutdown) {
        // 阻塞生产者线程
        pthread_cond_wait(&pool->notFull, &pool->mutexPool);
    }

    // 阻塞结束，线程池是否还存活
    if (pool->shutdown) {
        pthread_mutex_unlock(&pool->mutexPool);
        return;
    }

    // 添加任务
    pool->taskQ[pool->queueRear].function = func;
    pool->taskQ[pool->queueRear].args = arg;
    pool->queueRear = (pool->queueRear + 1) % pool->queueCapacity;      // 环形队列移动队尾
    pool->queueSize++;

    pthread_cond_signal(&pool->notEmpty);                               // 唤醒工作线程
    pthread_mutex_unlock(&pool->mutexPool);

    return;
}

int threadPoolBusyNum(ThreadPool* pool) {
    pthread_mutex_lock(&pool->mutexBusy);
    int busyNum = pool->busyNum;
    pthread_mutex_unlock(&pool->mutexBusy);
    return busyNum;
}

int threadPoolAliveNum(ThreadPool* pool) {
    pthread_mutex_lock(&pool->mutexPool);
    int aliveNum = pool->liveNum;
    pthread_mutex_unlock(&pool->mutexPool);
    return aliveNum;
}

int threadPoolDestory(ThreadPool* pool) {
    if (pool == NULL) {
        return -1;
    }

    // 关闭线程
    pool->shutdown = 1;

    // 阻塞回收管理者线程
    pthread_join(pool->managerID, NULL);

    // 唤醒阻塞的消费者线程
    for (int i = 0; i < pool->liveNum; ++i) {
        // 唤醒worker线程，引导其自杀
        pthread_cond_signal(&pool->notEmpty);
    }

    // 释放堆内存
    if (pool->taskQ != NULL) {
        free(pool->taskQ);
    }
    if (pool->threaIDs != NULL) {
        free(pool->threaIDs);
    }

    // 销毁条件变量
    pthread_mutex_destroy(&pool->mutexBusy);
    pthread_mutex_destroy(&pool->mutexPool);
    pthread_cond_destroy(&pool->notEmpty);
    pthread_cond_destroy(&pool->notFull);

    free(pool);
    pool = NULL;

    printf("thread pool has destoryed.\n");

    return 1;
}